/////////////////////////////////////////////////////////////////////////////
// Original code from libhdfs3. Copyright (c) 2013 - 2014, Pivotal Inc.
// All rights reserved. Author: Zhanwei Wang
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//



#include <kmhdfs/common/big_endian.h>
#include <kmhdfs/common/exception.h>
#include <kmhdfs/common/exception_internal.h>
#include <kmhdfs/client/packet.h>
#include <kmhdfs/client/packet_header.h>

namespace Hdfs {
namespace Internal {

Packet::Packet() :
    lastPacketInBlock(false), syncBlock(false), checksumPos(0), checksumSize(0),
    checksumStart(0), dataPos(0), dataStart(0), headerStart(0), maxChunks(
        0), numChunks(0), offsetInBlock(0), seqno(HEART_BEAT_SEQNO) {
    buffer.resize(PacketHeader::GetPkgHeaderSize());
}

Packet::Packet(int pktSize, int chunksPerPkt, int64_t offsetInBlock,
               int64_t seqno, int checksumSize) :
    lastPacketInBlock(false), syncBlock(false), checksumSize(checksumSize), headerStart(0),
    maxChunks(chunksPerPkt), numChunks(0), offsetInBlock(offsetInBlock), seqno(seqno), buffer(pktSize) {
    checksumPos = checksumStart = PacketHeader::GetPkgHeaderSize();
    dataPos = dataStart = checksumStart + chunksPerPkt * checksumSize;
    assert(dataPos >= 0);
}

void Packet::reset(int pktSize, int chunksPerPkt, int64_t offsetInBlock,
                   int64_t seqno, int checksumSize) {
    lastPacketInBlock = false;
    syncBlock = false;
    this->checksumSize = checksumSize;
    headerStart = 0;
    maxChunks = chunksPerPkt;
    numChunks = 0;
    this->offsetInBlock = offsetInBlock;
    this->seqno = seqno;
    checksumPos = checksumStart = PacketHeader::GetPkgHeaderSize();
    dataPos = dataStart = checksumStart + chunksPerPkt * checksumSize;

    if (pktSize > static_cast<int>(buffer.size())) {
        buffer.resize(pktSize);
    }

    assert(dataPos >= 0);
}

void Packet::addChecksum(uint32_t checksum) {
    if (checksumPos + static_cast<int>(sizeof(uint32_t)) > dataStart) {
        THROW(HdfsIOException,
              "Packet: failed to add checksum into packet, checksum is too large");
    }

    WriteBigEndian32ToArray(checksum, buffer.data() + checksumPos);
    checksumPos += checksumSize;
}

void Packet::addData(const char * buf, int size) {
    if (size + dataPos > static_cast<int>(buffer.size())) {
        THROW(HdfsIOException,
              "Packet: failed add data to packet, packet size is too small");
    }

    memcpy(buffer.data() + dataPos, buf, size);
    dataPos += size;
    assert(dataPos >= 0);
}

void Packet::setSyncFlag(bool sync) {
    syncBlock = sync;
}

void Packet::increaseNumChunks() {
    ++numChunks;
}

bool Packet::isFull() {
    return numChunks >= maxChunks;
}

bool Packet::isHeartbeat() {
    return HEART_BEAT_SEQNO == seqno;
}

void Packet::setLastPacketInBlock(bool lastPacket) {
    lastPacketInBlock = lastPacket;
}

int Packet::getDataSize() {
    return dataPos - dataStart;
}

int64_t Packet::getLastByteOffsetBlock() {
    assert(offsetInBlock >= 0 && dataPos >= dataStart);
    assert(dataPos - dataStart <= maxChunks * static_cast<int>(buffer.size()));
    return offsetInBlock + dataPos - dataStart;
}

const ConstPacketBuffer Packet::getBuffer() {
    /*
     * Once this is called, no more data can be added to the packet.
     * This is called only when the packet is ready to be sent.
     */
    int dataLen = dataPos - dataStart;
    int checksumLen = checksumPos - checksumStart;

    if (checksumPos != dataStart) {
        /*
         * move the checksum to cover the gap.
         * This can happen for the last packet.
         */
        memmove(&buffer[dataStart - checksumLen], buffer.data() + checksumStart,
                checksumLen);
        headerStart = dataStart - checksumPos;
        checksumStart += dataStart - checksumPos;
        checksumPos = dataStart;
    }

    assert(dataPos >= 0);
    int pktLen = dataLen + checksumLen;
    PacketHeader header(pktLen + sizeof(int32_t)
                        /* why we add 4 bytes? Because the server will reduce 4 bytes. -_-*/
                        , offsetInBlock, seqno, lastPacketInBlock, dataLen);
    header.writeInBuffer(buffer.data() + headerStart,
                         PacketHeader::GetPkgHeaderSize());
    return ConstPacketBuffer(buffer.data() + headerStart,
                             PacketHeader::GetPkgHeaderSize() + pktLen);
}

}
}
